package com.fangcang.elasticsearch;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import com.fangcang.dao.FailLogEntity;
import com.fangcang.dao.LogEntity;
import com.fangcang.utils.LocalDateUtils;

/**
 * 去哪儿变价推送统计
 * 
 * @author fangcang
 *
 */
@Service
public class QunarPushStatisticService {

	private Logger logger = Logger.getLogger(QunarPushStatisticService.class);

	private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 50, 10000, TimeUnit.SECONDS,
			new LinkedBlockingDeque<Runnable>(500));

	public List<Map<String, Object>> getSearchResult(Map<String, String> queryMap) {

		return null;
	}

	public QueryBuilder getQueryBuilder(EsQueryParam queryParam, Map<String, Object> queryMap) {
		long fromTime = LocalDateUtils.getLongTime(queryParam.getFromTime());
		long endTime = LocalDateUtils.getLongTime(queryParam.getEndTime());
		
		BoolQueryBuilder query = QueryBuilders.boolQuery()
				.must(QueryBuilders.rangeQuery(queryParam.getRangeField()).gte(fromTime).lt(endTime));
		if (!queryMap.isEmpty()) {
			Set<Entry<String, Object>> entrySet = queryMap.entrySet();
			entrySet.forEach(entry -> {
				query.must(QueryBuilders.termQuery(entry.getKey(), entry.getValue()));
			});
		}
		return query;
	}

	public TransportClient getClient() throws UnknownHostException {
		@SuppressWarnings("resource")
		TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
				.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("127.0.0.1"), 9300));
		return client;
	}

	public void search(EsQueryParam queryParam, Map<String, Object> queryMap) throws Exception {
		TransportClient client = getClient();
		logger.info("formTime：" + queryParam.getFromTime() + "-------endTime：" + queryParam.getEndTime());
		// 查询上一个十分钟内的数据：如当前时间为15:25:45，则查询15:10:00-15:19:59

		QueryBuilder query = getQueryBuilder(queryParam, queryMap);

		String scrollId = "";
		SearchResponse scrollResponse = scrollSearch(queryParam.getIndex(), client, query);
		while (scrollResponse != null) {
			SearchHit hits[] = scrollResponse.getHits().getHits();
			int length = hits == null ? 0 : hits.length;
			logger.info("scrollId：" + scrollResponse.getScrollId() + "，查询" + length + "条记录");
			List<Map<String, Object>> datas = new ArrayList<Map<String, Object>>(length);
			for (SearchHit hit : hits) {
				datas.add(hit.getSource());
			}
			scrollId = scrollResponse.getScrollId();
			if (scrollId == null || "".equals(scrollId) || datas.isEmpty()) {
				return;
			}
			threadPool.execute(new QunarPushTask(datas,
					queryParam.getFromTime().format(DateTimeFormatter.ofPattern("yyyyMMddHHmm"))));
			scrollResponse = client.prepareSearchScroll(scrollId).setScroll(TimeValue.timeValueMinutes(8)).execute()
					.actionGet();
		}
		if (scrollId != null && !"".equals(scrollId)) {
			ClearScrollRequest clearRequest = new ClearScrollRequest();
			clearRequest.addScrollId(scrollId);
			client.clearScroll(clearRequest);
		}
		client.close();
	}

	private SearchResponse scrollSearch(String index, TransportClient client, QueryBuilder query) {
		SearchResponse scrollResponse = client.prepareSearch(index).addSort(SortBuilders.fieldSort("startTime"))
				.setSize(EsConstants.SCROLL_SIZE) // 返回的数据量就是参数中指定的数据量
				.setScroll(TimeValue.timeValueMinutes(1)).setQuery(query).execute().actionGet();
		if (scrollResponse == null || scrollResponse.getHits() == null || scrollResponse.getHits().getHits() == null
				|| scrollResponse.getHits().getHits().length == 0) {
			return null;
		}
		return scrollResponse;
	}

	public List<String> getMultiFieldSum(EsQueryParam queryParam, Map<String, Object> queryMap, Map<String, String> termMap) {
		if (termMap == null || termMap.isEmpty()) {
			return null;
		}
		TransportClient client = null;
		try {
			client = getClient();

			SearchRequestBuilder sbuilder = client.prepareSearch(queryParam.getIndex());
			//添加分组条件
			sbuilder.addAggregation(getAggregation(termMap));
			//添加查询条件
			sbuilder.setQuery(getQueryBuilder(queryParam, queryMap));

			SearchResponse responses = sbuilder.execute().actionGet();
			// 得到这个分组的数据集合
			Terms terms = responses.getAggregations().get("sum");
			for (int i = 0; i < terms.getBuckets().size(); i++) {
				String id = terms.getBuckets().get(i).getKey().toString();// id
				Long sum = terms.getBuckets().get(i).getDocCount();// 数量
				System.out.println("key==" + id + "------sum==" + sum);
			}
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}
		return null;
	}
	
	/**
	 * 查询单列分组数量
	 * @param queryParam
	 * @param queryMap
	 * @param termMap
	 * @return
	 */
	public List<String> getFieldCount(EsQueryParam queryParam, Map<String, Object> queryMap, Map<String, String> termMap) throws Exception{
		if (termMap == null || termMap.isEmpty()) {
			return null;
		}
		TransportClient client = null;
		try {
			client = getClient();

			SearchRequestBuilder sbuilder = client.prepareSearch(queryParam.getIndex());
			//添加分组条件
			sbuilder.addAggregation(getAggregation(termMap));
			//添加查询条件
			sbuilder.setQuery(getQueryBuilder(queryParam, queryMap));

			SearchResponse responses = sbuilder.execute().actionGet();
			// 得到这个分组的数据集合
			Terms terms = responses.getAggregations().get("count");
			for (int i = 0; i < terms.getBuckets().size(); i++) {
				String id = terms.getBuckets().get(i).getKey().toString();// id
				Long sum = terms.getBuckets().get(i).getDocCount();// 数量
				System.out.println("key==" + id + "------sum==" + sum);
			}
		} catch (UnknownHostException e) {
			logger.error("UnknownHostException：", e);
			throw e;
		} finally {
			if(null != client) {
				client.close();
			}
		}
		return null;
	}
	
	
	/**
	 * 按商户和失败原因统计
	 * @param queryParam
	 * @param queryMap
	 * @param termMap
	 * @return
	 * @throws Exception
	 */
	public List<FailLogEntity> getMerchantFailCount(EsQueryParam queryParam, Map<String, Object> queryMap, Map<String, String> termMap) throws Exception{
		List<FailLogEntity> failList = new ArrayList<FailLogEntity>();
		if (termMap == null || termMap.isEmpty()) {
			return failList;
		}
		TransportClient client = null;
		try {
			client = getClient();

			SearchRequestBuilder sbuilder = client.prepareSearch(queryParam.getIndex());
			//添加分组条件
			sbuilder.addAggregation(getAggregation(termMap));
			//添加查询条件
			sbuilder.setQuery(getQueryBuilder(queryParam, queryMap));

			SearchResponse responses = sbuilder.execute().actionGet();
			// 得到这个分组的数据集合
			Terms terms = responses.getAggregations().get("count");
			for (int i = 0; i < terms.getBuckets().size(); i++) {
				
				String id = terms.getBuckets().get(i).getKey().toString();// 商户编号
//				Long sum = terms.getBuckets().get(i).getDocCount();// 数量
				Terms sumTerms = terms.getBuckets().get(i).getAggregations().get("subCount");
				for(Terms.Bucket sumTerm :  sumTerms.getBuckets()) {
					FailLogEntity result = new FailLogEntity();
					result.setAppName(queryParam.getAppName());
					result.setBizName(queryParam.getBizName());
					result.setFailType(sumTerm.getKeyAsString());
					result.setFailNum(sumTerm.getDocCount());
					result.setMerchantCode(id);
					result.setStartTime(LocalDateUtils.localDateTime2Date(queryParam.getFromTime()));
					result.setEndTime(LocalDateUtils.localDateTime2Date(queryParam.getEndTime()));
					result.setCreatTime(Calendar.getInstance().getTime());
//					System.out.println("key="+id + ":" + sumTerm.getKeyAsString() +  "-------sum="+sumTerm.getDocCount());
					failList.add(result);
				}
//				System.out.println("key==" + id + "------sum==" + sum);
			}
		} catch (UnknownHostException e) {
			logger.error("UnknownHostException：", e);
			throw e;
		} finally {
			if(null != client) {
				client.close();
			}
		}
		return failList;
	}
	
	
	/**
	 * 按商户查询成功和失败数(商户、bizName、logType)
	 * @param queryParam
	 * @param queryMap
	 * @param termMap
	 * @return
	 * @throws Exception
	 */
	public List<LogEntity> getMerchantCount(EsQueryParam queryParam) throws Exception{
		List<LogEntity> logList = new ArrayList<LogEntity>();
		TransportClient client = null;
		try {
			client = getClient();

			SearchRequestBuilder sbuilder = client.prepareSearch(queryParam.getIndex());
			//添加分组条件
			TermsAggregationBuilder termsBuilder = AggregationBuilders.terms("merchantCount").field("merchantCode.keyword");
			TermsAggregationBuilder termsBuilder1 = AggregationBuilders.terms("bizCount").field("bizName.keyword");
			TermsAggregationBuilder termsBuilder2 = AggregationBuilders.terms("logTypeCount")	.field("logType.keyword");
			
			sbuilder.addAggregation(termsBuilder.subAggregation(termsBuilder1.subAggregation(termsBuilder2)));
			//添加查询条件
			long fromTime = LocalDateUtils.getLongTime(queryParam.getFromTime());
			long endTime = LocalDateUtils.getLongTime(queryParam.getEndTime());
			sbuilder.setQuery(QueryBuilders.boolQuery()
					.must(QueryBuilders.rangeQuery(queryParam.getRangeField()).gte(fromTime).lt(endTime)));

			SearchResponse responses = sbuilder.execute().actionGet();
			// 取商户分组
			Terms terms = responses.getAggregations().get("merchantCount");
			if(null == terms || CollectionUtils.isEmpty(terms.getBuckets())) {
				return logList;
			}
			for (int i = 0; i < terms.getBuckets().size(); i++) {
				Terms.Bucket buc = terms.getBuckets().get(i);
				String merchantCode = buc.getKey().toString();		// 商户编号
				//取业务分组
				Terms bizTerms = buc.getAggregations().get("bizCount");
				if(null == bizTerms || CollectionUtils.isEmpty(bizTerms.getBuckets()) ) {
					return logList;
				}
				for(Terms.Bucket sumTerm :  bizTerms.getBuckets()) {
					String bizName =  sumTerm.getKeyAsString();
					Terms logTypeTerms = sumTerm.getAggregations().get("logTypeCount");
					if(null == logTypeTerms || CollectionUtils.isEmpty(logTypeTerms.getBuckets()) ) {
						return logList;
					}
					LogEntity logEntity = new LogEntity();
					logEntity.setAppName(queryParam.getAppName());
					logEntity.setBizName(bizName);
					logEntity.setMerchantCode(merchantCode);
					logEntity.setStartTime(LocalDateUtils.localDateTime2Date(queryParam.getFromTime()));
					logEntity.setEndTime(LocalDateUtils.localDateTime2Date(queryParam.getEndTime()));
					logEntity.setCreatTime(Calendar.getInstance().getTime());
					
					for(Terms.Bucket logTerm : logTypeTerms.getBuckets()) {
						String logType = logTerm.getKeyAsString() ;
						if("end".equals(logType)) {
							logEntity.setSuccessNum(Long.valueOf(sumTerm.getDocCount()).intValue());
							//System.out.println("merchantCode="+merchantCode + "---bizName=" +bizName  +  "---logType="+logType+"--success="+sumTerm.getDocCount());
						}else if("error".equals(logType)){
							logEntity.setFailNum(Long.valueOf(sumTerm.getDocCount()).intValue());
							//System.out.println("merchantCode="+merchantCode + "---bizName=" +bizName  +  "---logType="+logType+"--fail="+sumTerm.getDocCount());
						}
					}
					logList.add(logEntity);
				}
			}
		} catch (UnknownHostException e) {
			logger.error("UnknownHostException：", e);
			throw e;
		} finally {
			if(null != client) {
				client.close();
			}
		}
		return logList;
	}

	/**
	 * 获取分组条件
	 * @param termMap
	 * @return
	 */
	private AggregationBuilder getAggregation(Map<String, String> termMap) {
		AggregationBuilder firstTermsBuilder = null;
		int termsNum = 0;
		for (Entry<String, String> entry : termMap.entrySet()) {
			TermsAggregationBuilder termsBuilder = AggregationBuilders.terms(entry.getValue())
					.field(entry.getKey());
			if (termsNum == 0) {
				firstTermsBuilder = termsBuilder;
			} else {
				firstTermsBuilder.subAggregation(termsBuilder);
			}
			termsNum++;
		}
		return firstTermsBuilder;
	}
	
	public List<String> getMultiFieldCount(EsQueryParam queryParam, Map<String, Object> queryMap, Map<String, String> termMap) {
		if (termMap == null || termMap.isEmpty()) {
			return null;
		}
		TransportClient client = null;
		try {
			client = getClient();

			SearchRequestBuilder sbuilder = client.prepareSearch(queryParam.getIndex());

			sbuilder.addAggregation(getAggregation(termMap));
			
			sbuilder.setQuery(getQueryBuilder(queryParam, queryMap));

			SearchResponse responses = sbuilder.execute().actionGet();
			// 得到这个分组的数据集合
			Terms terms = responses.getAggregations().get("count");
			for (int i = 0; i < terms.getBuckets().size(); i++) {
				String id = terms.getBuckets().get(i).getKeyAsString();// id
//				Long count = terms.getBuckets().get(i).getDocCount();// 数量
				Terms sumTerms = terms.getBuckets().get(i).getAggregations().get("subCount");
				for(Terms.Bucket sumTerm :  sumTerms.getBuckets()) {
					System.out.println("key="+id + ":" + sumTerm.getKeyAsString() +  "-------sum="+sumTerm.getDocCount());
				}
			}
		} catch (UnknownHostException e) {
			e.printStackTrace();
		} finally {
			if(null != client) {
				client.close();
			}
		}
		return null;
	}

}
